package com.atguigu.gmall.realtime.utils;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 维度流sink函数
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixConnection");
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);

    }

    @Override
    public void invoke(JSONObject jsonObject, Context context) throws Exception {
        // TODO: 获取输出的Phoenix
        String tableName = jsonObject.getString("sink_table").toUpperCase();
        // 获取data数据
        JSONObject datajson = jsonObject.getJSONObject("data");
        if (datajson != null && datajson.size() > 0) {
            String upsertSql = genUpsertSql(tableName, datajson);
            PreparedStatement ps = null;

            try {
                ps = connection.prepareStatement(upsertSql);
                ps.executeUpdate();
                connection.commit();
                System.out.println("向Phoenix中插入数据的Sql:" + upsertSql);
            } catch (SQLException e) {
                e.printStackTrace();
                throw new RuntimeException("向Phoenix中插入数据失败");
            } finally {
                if (ps != null) {
                    ps.close();
                }
            }


        }


    }

    // 拼接插入到Phonenix表的SQL
    private String genUpsertSql(String tableName, JSONObject datajson) {
        String upsertSql = "upsert into " + GmallConfig.HBASE_SCHEMA
                + "." + tableName
                + " (" + StringUtils.join(datajson.keySet(), ",") + ")" +
                "values (\' " + StringUtils.join(datajson.values(), "\',\'") + "\')";


        return upsertSql;
    }
}
